package com.why.cachecloud.schedule.jobs;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.google.common.collect.Lists;
import com.why.cachecloud.configuration.constant.AppStatusEnum;
import com.why.cachecloud.configuration.constant.InstanceStatusEnum;
import com.why.cachecloud.dao.InstanceBigKeyDao;
import com.why.cachecloud.dao.InstanceDao;
import com.why.cachecloud.entity.AppDesc;
import com.why.cachecloud.entity.InstanceInfo;
import com.why.cachecloud.entity.WarnConfig;
import com.why.cachecloud.service.alert.WarnConfigService;
import com.why.cachecloud.service.alert.WarnType;
import com.why.cachecloud.service.custom.alert.DingTalkComponent;
import com.why.cachecloud.service.redis.RedisCenter;
import com.why.cachecloud.service.task.constant.InstanceRoleEnum;
import com.why.cachecloud.service.task.constant.RedisDataStructureTypeEnum;
import com.why.cachecloud.service.task.entity.InstanceBigKey;
import com.why.cachecloud.util.ConstUtils;
import com.why.cachecloud.util.StringUtil;
import com.why.cachecloud.web.enums.BooleanEnum;
import com.why.cachecloud.web.service.AppService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.ScanParams;
import redis.clients.jedis.ScanResult;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static com.why.cachecloud.util.ConstUtils.CACHE_REDIS_SENTINEL;


/**
 * @Author: why
 * @Date: 2020/6/9 15:53
 */
@Slf4j
@Component
public class ScanBigKeyService {

    @Resource
    protected AppService appService;

    @Resource
    protected InstanceDao instanceDao;

    @Autowired
    protected RedisCenter redisCenter;

    @Resource
    private InstanceBigKeyDao instanceBigKeyDao;

    @Resource
    private DingTalkComponent dingTalkComponent;

    @Resource
    private WarnConfigService warnConfigService;

    private final int SCAN_COUNT = 100;
    private Map<String, Long> configLength = new HashMap<>();

    public void scanBigKey() {
        begin();
        String jobParam = ""; //XxlJobHelper.getJobParam();
        Set<Long> appointAppId = Collections.emptySet();
        if (StringUtils.isNotBlank(jobParam)) {
            appointAppId = Arrays.stream(jobParam.split(",")).map(Long::valueOf).collect(Collectors.toSet());
        }

        List<Integer> statusList = Collections.singletonList(AppStatusEnum.STATUS_PUBLISHED.getStatus());
        List<AppDesc> appList = appService.queryListByStatus(statusList);
        if (CollectionUtils.isEmpty(appList)) {
            log.info("扫描大KEY结束,可用的实例列表为空");
            return;
        }

        Set<AppDesc> appDescSet = new HashSet<>();
        List<InstanceBigKey> allBigKeyList = new ArrayList<>();
        for (AppDesc appDesc : appList) {
            long appId = appDesc.getAppId();
            if (!appointAppId.isEmpty() && !appointAppId.contains(appId)) {
                continue;
            }
            List<InstanceInfo> instanceInfos = instanceDao.getEffectiveInstListByAppId(appId);
            List<InstanceInfo> infoList = instanceInfos.stream().filter(i -> i.getStatus() == InstanceStatusEnum.GOOD_STATUS.getStatus()).collect(Collectors.toList());
            if (CollectionUtils.isEmpty(infoList)) {
                log.info("scanBigKey 根据appId:{},查询出来的可用实例列表为空", appId);
                continue;
            }
            for (InstanceInfo instanceInfo : infoList) {
                if (instanceInfo.getType() == CACHE_REDIS_SENTINEL) {
                    continue;
                }
                BooleanEnum isMaster = redisCenter.isMaster(appId, instanceInfo.getIp(), instanceInfo.getPort());
                if (!BooleanEnum.TRUE.equals(isMaster)) {
                    continue;
                }
                List<InstanceBigKey> stringLongMap = scanBigKey(appId, instanceInfo.getId(), instanceInfo.getIp(), instanceInfo.getPort());
                if (!stringLongMap.isEmpty()) {
                    allBigKeyList.addAll(stringLongMap);
                    appDescSet.add(appDesc);
                }
            }
        }
        if (allBigKeyList.isEmpty()) {
            log.info("扫描大KEY结束,扫描到的记录结果为空");
            return;
        }
        String appIds = appDescSet.stream().map(AppDesc::getAppId).map(Object::toString).collect(Collectors.joining(","));
        log.info("扫描大KEY结果,涉及的应用为:{}", appIds);

        List<List<InstanceBigKey>> partition = Lists.partition(allBigKeyList, 500);

        for (List<InstanceBigKey> instanceBigKeys : partition) {
            instanceBigKeyDao.batchSave(instanceBigKeys);
        }
        dingTalkComponent.sendDingTalk(WarnType.bigKey, appDescSet);
        end();
    }

    /**
     * @param appId  应用ID
     * @param instId 实例ID
     * @param host   IP
     * @param port   端口
     * @return 大key集合对象
     */
    public List<InstanceBigKey> scanBigKey(long appId, long instId, String host, int port) {
        /**
         * 扫描bigkey，计时开始*/
        long startTime = System.currentTimeMillis();
        Jedis jedis = null;
        List<InstanceBigKey> retList = new ArrayList<>();
        try {
            jedis = redisCenter.getJedis(appId, host, port);
            long dbSize = jedis.dbSize();
            if (dbSize == 0) {
                log.info("{} {}:{} dbsize is {}", appId, host, port, dbSize);
                return retList;
            }
            log.info("{} {}:{} total key is {} ", appId, host, port, dbSize);

            // scan参数
            byte[] cursor = "0".getBytes(StandardCharsets.UTF_8);
            ScanParams scanParams = new ScanParams().count(SCAN_COUNT);

            long count = 0;
            int totalSplit = 10;
            int curSplit = 1;

            while (true) {
                try {
                    ScanResult<byte[]> scanResult = jedis.scan(cursor, scanParams);
                    cursor = scanResult.getCursorAsBytes();
                    List<byte[]> keyList = scanResult.getResult();

                    // 使用pipeline获取type
                    Pipeline pipeline = jedis.pipelined();
                    for (byte[] bytes : keyList) {
                        pipeline.type(bytes);
                    }

                    List<Object> typeList = pipeline.syncAndReturnAll();
                    Pipeline pipeline2 = jedis.pipelined();
                    for (int i = 0; i < keyList.size(); i++) {
                        String type = String.valueOf(typeList.get(i));
                        String key = new String(keyList.get(i));
                        if ("none".equals(type)) {
                            log.warn("none.equals(type),key:{},type:{}", key, type);
                            continue;
                        }
                        if (RedisDataStructureTypeEnum.string.getValue().equals(type)) {
                            pipeline2.strlen(key);
                        } else if (RedisDataStructureTypeEnum.hash.getValue().equals(type)) {
                            pipeline2.hlen(key);
                        } else if (RedisDataStructureTypeEnum.list.getValue().equals(type)) {
                            pipeline2.llen(key);
                        } else if (RedisDataStructureTypeEnum.set.getValue().equals(type)) {
                            pipeline2.scard(key);
                        } else if (RedisDataStructureTypeEnum.zset.getValue().equals(type)) {
                            pipeline2.zcard(key);
                        }
                    }

                    List<Object> lengthList = pipeline2.syncAndReturnAll();

                    List<InstanceBigKey> instanceBigKeyList = new ArrayList<>(lengthList.size());
                    for (int i = 0; i < lengthList.size(); i++) {
                        Object lenObj = lengthList.get(i);
                        if (!(lenObj instanceof Long)) {
                            continue;
                        }
                        long length = (long) lenObj;
                        String type = String.valueOf(typeList.get(i));
                        byte[] key = keyList.get(i);
                        boolean isBigKey = checkIsBigKey(type, length);
                        if (!isBigKey) {
                            continue;
                        }
                        InstanceBigKey instanceBigKey = new InstanceBigKey();
                        instanceBigKey.setAppId(appId);

                        instanceBigKey.setInstanceId(instId);
                        instanceBigKey.setAuditId(0);
                        instanceBigKey.setIp(host);
                        instanceBigKey.setPort(port);
                        instanceBigKey.setRole(InstanceRoleEnum.MASTER.getRole());
                        instanceBigKey.setBigKey(new String(key));
                        instanceBigKey.setLength(length);
                        instanceBigKey.setType(type);
                        instanceBigKey.setCreateTime(new Date());
                        instanceBigKeyList.add(instanceBigKey);
                    }

                    if (CollectionUtils.isNotEmpty(instanceBigKeyList)) {
                        retList.addAll(instanceBigKeyList);
                    }

                    count += keyList.size();
                    if (count > dbSize / totalSplit * curSplit) {
                        log.info("{} {}:{} has already anlysis {}% {} key ", appId, host, port, curSplit * 10, count);
                        curSplit++;
                    }
                    // @TODO暂时写死
                    TimeUnit.MILLISECONDS.sleep(2);
                } catch (Exception e) {
                    log.error("大key检测出现异常", e);
                } finally {
                    //防止无限循环
                    if (Arrays.equals("0".getBytes(StandardCharsets.UTF_8), cursor)) {
                        break;
                    }
                }
            }
            long cost = System.currentTimeMillis() - startTime;
            log.info("{} {}:{} instanceBigKeyTask successfully, cost time is {} ms, total key is {}", appId, host, port, cost, count);
        } catch (RuntimeException e) {
            log.error("大key检测出现异常", e);
        } catch (Exception e) {
            log.error("redis-cli -h {} -p {} admin auth error", host, port);
            log.error("instanceBigKeyTask appId {} {}:{}  error:" + e.getMessage(), appId, host, port, e);
        } finally {
            if (jedis != null) {
                jedis.close();
            }
        }
        return retList;
    }

    /**
     * 按照类型判断是否为bigkey
     *
     * @param type
     * @param length
     * @return
     */
    private boolean checkIsBigKey(String type, long length) {
        return length > configLength.get(type);
    }

    private void begin() {
        if (configLength.isEmpty()) {
            WarnConfig warnConfig = warnConfigService.queryConfig(0, WarnType.bigKey);
            if (warnConfig != null) {
                String parameter = warnConfig.getParameter();
                if (!StringUtil.isBlank(parameter)) {
                    Map<String, Long> stringLongMap = JSON.parseObject(parameter, new TypeReference<Map<String, Long>>() {
                    });
                    if (!stringLongMap.isEmpty()) {
                        configLength.putAll(stringLongMap);
                    }
                }
            }
            configLength.putIfAbsent(RedisDataStructureTypeEnum.string.getValue(), Long.valueOf(String.valueOf(ConstUtils.STRING_MAX_LENGTH)));
            configLength.put(RedisDataStructureTypeEnum.hash.getValue(), Long.valueOf(String.valueOf(ConstUtils.HASH_MAX_LENGTH)));
            configLength.put(RedisDataStructureTypeEnum.list.getValue(), Long.valueOf(String.valueOf(ConstUtils.LIST_MAX_LENGTH)));
            configLength.put(RedisDataStructureTypeEnum.set.getValue(), Long.valueOf(String.valueOf(ConstUtils.SET_MAX_LENGTH)));
            configLength.put(RedisDataStructureTypeEnum.zset.getValue(), Long.valueOf(String.valueOf(ConstUtils.ZSET_MAX_LENGTH)));
        }
    }

    private void end() {
        configLength.clear();
    }
}
